package com.lanou.quanttradedata.stock.mapreduce.reducer;

import com.lanou.quanttradedata.QuantTradeDataApplication;
import com.lanou.quanttradedata.stock.javabean.BiasInfo;
import com.lanou.quanttradedata.stock.mapper.BiasInfoMapper;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Date;

/**
 * 这里是业务层的减量器，
 * 可以对数据进行业务筛选计算，在映射类处理后，将映射类映射合并的数据进行进一步业务计算，
 * 在此计算出最终结果，保存到HDFS中，并将结果保存数据库
 */
public class BiasCalculateReducer extends Reducer<Text,Text,Text,Text> {
    //mapper层注入
    private BiasInfoMapper biasInfoMapper = (BiasInfoMapper) QuantTradeDataApplication.context.getBean("biasInfoMapper");
    private SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
    @Override
    protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {
        //1、将values转储到biasI
        // nfos中
        ArrayList<BiasInfo> biasInfos = new ArrayList<>(10);
        //     获取7天前日期
        Integer firstDay = Integer.valueOf(sdf.format(new Date(System.currentTimeMillis()-1000L*60*60*24*7)));
        //2、遍历values，遍历该股票每天的日线数据
        //收盘价累加和
        float sum = 0.0F;
        //保存每个遍历的收盘价，用于乖离率计算
        ArrayDeque<Float> stockCloseDeque = new ArrayDeque<>();
        for (Text value : values) {
            String[] line = value.toString().split(" ");
            //交易日期
            String tradedate = line[0];
            //收盘价
            float stockclose = Float.parseFloat(line[2]);
            //维护收盘价累加和
            sum+=stockclose;
            stockCloseDeque.addLast(stockclose);
            //永远保持最多只有7个收盘价，这样如果要计算乖离率，除去今天，就是6日乖离率
            if(stockCloseDeque.size()>=8){
                sum-=stockCloseDeque.removeFirst();
            }
            //3、对交易日期进行判断，如果交易日期符合要求，并且今天至少是第7天，则计算乖离率并保存
            if(Integer.valueOf(tradedate)>=firstDay && stockCloseDeque.size()>=7){
                //计算6日平均价 = (累加和-今日收盘价)/6
                float avgClose = (sum-stockclose)/(stockCloseDeque.size()-1);
                //计算乖离率
                float bias= (stockclose-avgClose)/avgClose;
                //开盘价
                float stockopen = Float.parseFloat(line[1]);
                //昨日收盘价
                float preclose = Float.parseFloat(line[3]);
                //最高价
                float high = Float.parseFloat(line[4]);
                //最低价
                float low = Float.parseFloat(line[5]);
                //4、计算好乖离率之后，将当天内容保存 乖离率对象数据的集合，并向hdfs输出
                biasInfos.add(
                        new BiasInfo(null,key.toString(),tradedate,stockopen,stockclose,preclose,high,low,bias)
                );
                context.write(key,new Text(
                        tradedate+" "
                        +stockopen+" "
                        +stockclose+" "
                        +preclose+" "
                        +high+" "
                        +low+" "
                        +bias
                ));
            }
        }
        //5、向数据库一次性保存数据
        biasInfoMapper.saveBiasInfos(biasInfos);
    }
}
